#include "aio_tcp.h"
#include "../global/head.h"
#include "../xos/head.h"
#include "../buf/head.h"
#include "aio_callback_thread.h"
#include "aio_impl.h"
#include "aio_data.h"
#include "aio_key.h"

/////////////////////////////////////////////////////////////////////////////////////////
// 
namespace xos
{
    aio_tcp::aio_tcp( aio_impl * pAioImpl ) : m_pAioImpl( pAioImpl )
    {
    }

    aio_tcp::~aio_tcp()
    {
    }

    // 
    // m_bData[0] : true or false for result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_init_server_ret( int nResult, DWORD dwBytes, aio_key * lpKey, aio_data * lpData )
    {
        int nRet = 0;

        msg_impl * lpMsg = lpData->m_lpMsgData;
        aio_key * lpAioKey = ( aio_key * )lpMsg->m_lpData[1];

        if( !lpMsg->m_bData[0] )
        {
            if( lpAioKey )
            {
                lpMsg->m_lpData[1] = 0;
                lpAioKey->release();
                lpAioKey = 0;
            }
        }

        m_pAioImpl->m_pCallbackThread->notify( lpMsg );

        return nRet;
    }

    // 
    // m_bData[0] : true or false for result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_init_server( int nResult, DWORD dwBytes, aio_key * lpKey, aio_data * lpData )
    {
        int nRet = 0;

        msg_impl * lpMsg = lpData->m_lpMsgData;
        void * lpUserKey = lpMsg->m_lpData[0];
        aio_key * lpAioKey = 0;
        tcp_impl * lpTcp = 0;
        socket_impl * lpSocket = 0;

        SOCKET hSocket = INVALID_SOCKET;
        sockaddr_in addr = { 0 };
        char szIp[100] = { 0 };
        DWORD dwIp = 0;
        aio_data * lpRetData = 0;

        if( 0 == nRet )
        {
            hSocket = ::WSASocket( AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED );

            if( INVALID_SOCKET == hSocket )
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            g_pXosImpl->create( i_xos::XOS_OBJ_TCP, ( void ** )&lpTcp );
            lpSocket = &lpTcp->m_socket;
            lpSocket->m_hSocket = hSocket;

            lpAioKey = m_pAioImpl->m_key_pool.GetItem();
            lpAioKey->init();
            lpAioKey->m_eState = aio_key::KEY_STATE_RUNNING;
            lpAioKey->m_lpUserKey = lpUserKey;
            lpAioKey->m_pHandle = lpTcp;

            lpMsg->m_lpData[1] = lpAioKey;
        }

        if( 0 == nRet )
        {
            nRet = lpSocket->get_ip_by_name( lpMsg->m_szStr[1], dwIp, szIp );
        }

        if( 0 == nRet )
        {
            addr.sin_port = htons( lpMsg->m_nData[1] );
            addr.sin_addr.s_addr = dwIp;
            addr.sin_family = AF_INET;

            if( SOCKET_ERROR == bind( hSocket, ( const sockaddr * )&addr, sizeof( sockaddr_in ) ) )
            {
                nRet = 2;
            }
        }

        if( 0 == nRet )
        {
            if( SOCKET_ERROR  == listen( hSocket, 10 ) )
            {
                nRet = 4;
            }
        }

        if( 0 == nRet )
        {
            if( CreateIoCompletionPort( ( HANDLE )hSocket, m_pAioImpl->m_hIocp, ( ULONG_PTR )lpAioKey, 0 ) != m_pAioImpl->m_hIocp )
            {
                nRet = 1;
            }
        }

        // 
        // create final result
        //
        lpRetData = m_pAioImpl->m_data_pool.GetItem();
        lpRetData->init();

        lpRetData->m_eAioType = i_aio::AIO_TCP_INIT_SERVER_RET;
        lpRetData->m_lpMsgData = lpMsg;
        lpMsg->m_nData[0] = lpRetData->m_eAioType;

        if( 0 == nRet )
        {
            memcpy( &lpSocket->m_addr, &addr, sizeof( addr ) );
            lpMsg->m_bData[0] = true;
        }
        else
        {
            lpMsg->m_bData[0] = false;
        }

        PostQueuedCompletionStatus( m_pAioImpl->m_hIocp, 0, 0, ( WSAOVERLAPPED * )&lpRetData->m_ov_ex );
        lpRetData = 0;

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_connect_ret( int nResult, DWORD dwBytes, aio_key * lpKey, aio_data * lpData )
    {
        int nRet = 0;

        msg_impl * lpMsg = lpData->m_lpMsgData;
        aio_key * lpAioKey = ( aio_key * )lpMsg->m_lpData[1];
        tcp_impl * lpTcp = 0;
        socket_impl * lpSocket = 0;
        xos_u16 usPort = 0;
        char szIp[100] = { 0 };

        if( !lpMsg->m_bData[0] || ( 0 != nResult ) )
        {
            nRet = 1;
        }

        if( 0 == nRet )
        {
            lpTcp = ( tcp_impl * )lpAioKey->m_pHandle;
            lpSocket = &lpTcp->m_socket;

            setsockopt( lpSocket->m_hSocket, 
                SOL_SOCKET, 
                SO_UPDATE_CONNECT_CONTEXT, 
                NULL, 
                0 );
        }

        if( 0 == nRet )
        {
            sockaddr_in addr = { 0 };
            int nLen = sizeof( addr );

            if( SOCKET_ERROR != getsockname( lpSocket->m_hSocket, ( sockaddr * )&addr, &nLen ) )
            {
                memcpy( &lpAioKey->m_local_addr, &addr, sizeof( addr ) );
                lstrcpyA( lpMsg->m_szStr[1], inet_ntoa( addr.sin_addr ) );
                lpMsg->m_nData[1] = ntohs( addr.sin_port );
            }
            else
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            sockaddr_in addr = { 0 };
            int nLen = sizeof( addr );

            if( SOCKET_ERROR != getpeername( lpSocket->m_hSocket, ( sockaddr * )&addr, &nLen ) )
            {
                memcpy( &lpAioKey->m_remote_addr, &addr, sizeof( addr ) );
                memcpy( &lpSocket->m_addr, &addr, sizeof( addr ) );
                lstrcpyA( lpMsg->m_szStr[2], inet_ntoa( addr.sin_addr ) );
                lpMsg->m_nData[2] = ntohs( addr.sin_port );
            }
            else
            {
                nRet = 1;
            }
        }

        if( 0 != nRet )
        {
            lpMsg->m_bData[0] = false;

            if( lpAioKey )
            {
                lpMsg->m_lpData[1] = 0;
                lpAioKey->release();
                lpAioKey = 0;
            }
        }

        m_pAioImpl->m_pCallbackThread->notify( lpMsg );

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_connect( int nResult, DWORD dwBytes, aio_key * lpKey, aio_data * lpData )
    {
        int nRet = 0;

        msg_impl * lpMsg = lpData->m_lpMsgData;
        void * lpUserKey = lpMsg->m_lpData[0];
        aio_key * lpAioKey = 0;
        tcp_impl * lpTcp = 0;
        socket_impl * lpSocket = 0;
        SOCKET hSocket = INVALID_SOCKET;

        xos::xos_u16 usPort = 0;
        char szIp[100] = { 0 };
        DWORD dwIp = 0;

        // 
        // prepare for result value
        // 
        aio_data * lpRetData = 0;
        lpRetData = m_pAioImpl->m_data_pool.GetItem();
        lpRetData->init();
        lpRetData->m_eAioType = i_aio::AIO_TCP_CONNECT_RET;
        lpRetData->m_lpMsgData = lpMsg;
        lpMsg->m_nData[0] = lpRetData->m_eAioType;

        if( 0 == nRet )
        {
            nRet = lpSocket->get_ip_by_name( lpMsg->m_szStr[2], dwIp, szIp );
        }

        if( 0 == nRet )
        {
            lpRetData->m_addr.sin_port = htons( lpMsg->m_nData[2] );
            lpRetData->m_addr.sin_addr.s_addr = dwIp;
            lpRetData->m_addr.sin_family = AF_INET;
        }

        if( 0 == nRet )
        {
            hSocket = ::WSASocket( AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED );

            if( INVALID_SOCKET == hSocket )
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            DWORD dwip = 0;
            lpSocket->get_ip_by_name( lpMsg->m_szStr[1], dwip, 0 );

            sockaddr_in l_addr = { 0 };
            l_addr.sin_port = htons( lpMsg->m_nData[1] );
            l_addr.sin_addr.s_addr = dwip;
            l_addr.sin_family = AF_INET;

            if( SOCKET_ERROR == bind( hSocket, ( const sockaddr * )&l_addr, sizeof( l_addr ) ) )
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            g_pXosImpl->create( i_xos::XOS_OBJ_TCP, ( void ** )&lpTcp );
            lpSocket = &lpTcp->m_socket;
            lpSocket->m_hSocket = hSocket;

            lpAioKey = m_pAioImpl->m_key_pool.GetItem();
            lpAioKey->init();
            lpAioKey->m_eState = aio_key::KEY_STATE_RUNNING;
            lpAioKey->m_lpUserKey = lpUserKey;
            lpAioKey->m_pHandle = lpTcp;

            lpMsg->m_lpData[1] = lpAioKey;
        }

        if( 0 == nRet )
        {
            if( CreateIoCompletionPort( ( HANDLE )hSocket, m_pAioImpl->m_hIocp, ( ULONG_PTR )lpAioKey, 0 ) != m_pAioImpl->m_hIocp )
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            lpMsg->m_bData[0] = true;

            if( !m_pAioImpl->m_lpConnectEx( hSocket, ( const sockaddr * )&lpRetData->m_addr, sizeof( sockaddr_in ), NULL, 0, NULL, ( WSAOVERLAPPED * )&lpRetData->m_ov_ex ) )
            {
                DWORD dwErr = WSAGetLastError();

                if( ERROR_IO_PENDING != dwErr )
                {
                    nRet = 1;
                }
            }
        }

        if( 0 != nRet )
        {
            lpMsg->m_bData[0] = false;
            PostQueuedCompletionStatus( m_pAioImpl->m_hIocp, 0, 0, ( WSAOVERLAPPED * )&lpRetData->m_ov_ex );
        }

        lpRetData = 0;

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // m_lpData[2] : ClientKey
    // 
    int aio_tcp::tcp_accept_ret( int nResult, DWORD dwBytes, aio_key * lpKey, aio_data * lpData )
    {
        int nRet = 0;

        msg_impl * lpMsg = lpData->m_lpMsgData;
        void * lpUserKey = lpMsg->m_lpData[0];
        aio_key * lpAioKey = ( aio_key * )lpMsg->m_lpData[1];
        tcp_impl * lpTcp = ( tcp_impl * )lpAioKey->m_pHandle;
        socket_impl * lpSocket = &lpTcp->m_socket;
        aio_key * lpClientKey = ( aio_key * )lpMsg->m_lpData[2];
        SOCKET hSocket = INVALID_SOCKET;
        buf_impl *& lpBuf = lpMsg->m_lpBuf[0];

        --lpAioKey->m_nRequestNum;

        if( !lpMsg->m_bData[0] || ( 0 != nResult ) )
        {
            nRet = 1;
        }

        if( 0 == nRet )
        {
            tcp_impl * pTcp = ( tcp_impl * )lpClientKey->m_pHandle;
            socket_impl * pSocket = &pTcp->m_socket;
            hSocket = pSocket->m_hSocket;

            DWORD dwS = sizeof( sockaddr_in );
            DWORD dwSize = dwS + 16;
            sockaddr_in * local = 0, * remote = 0;
            int l_size = dwS, r_size = dwS;

            m_pAioImpl->m_lpGetAcceptExSockaddrs( lpBuf->m_szBuf, 0, dwSize, dwSize, ( LPSOCKADDR * )&local, &l_size, ( LPSOCKADDR * )&remote, &r_size );
            memcpy( &lpClientKey->m_remote_addr, remote, dwS );
            memcpy( &lpClientKey->m_local_addr, local, dwS );
            memcpy( &pSocket->m_addr, remote, dwS );
            lstrcpyA( lpMsg->m_szStr[1], inet_ntoa( local->sin_addr ) );
            lpMsg->m_nData[1] = ntohs( local->sin_port );
            lstrcpyA( lpMsg->m_szStr[2], inet_ntoa( remote->sin_addr ) );
            lpMsg->m_nData[2] = ntohs( remote->sin_port );
        }

        if( 0 == nRet )
        {
            if( m_pAioImpl->m_hIocp != CreateIoCompletionPort( ( HANDLE )hSocket, m_pAioImpl->m_hIocp, ( ULONG_PTR )lpClientKey, 0 ) )
            {
                nRet = 1;
            }
        }

        if( 0 != nRet )
        {
            lpMsg->m_bData[0] = false;
            lpMsg->m_lpData[2] = 0;

            if( lpClientKey )
            {
                lpClientKey->release();
                lpClientKey = 0;
            }
        }

        lpBuf->release();
        lpBuf = 0;

        m_pAioImpl->m_pCallbackThread->notify( lpMsg );

        ///////////////////////////////////////////////////////////////////////////////////////////////
        // 
        // close connect
        // 

        if( ( aio_key::KEY_STATE_CLOSING == lpAioKey->m_eState ) && ( 0 == lpAioKey->m_nRequestNum ) )
        {
            helper_post_close( lpAioKey );
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // m_lpData[2] : ClientKey
    // 
    int aio_tcp::tcp_accept( int nResult, DWORD dwBytes, aio_key * lpKey, aio_data * lpData )
    {
        int nRet = 0;

        msg_impl * lpMsg = lpData->m_lpMsgData;
        void * lpUserKey = lpMsg->m_lpData[0];
        aio_key * lpAioKey = ( aio_key * )lpMsg->m_lpData[1];
        tcp_impl * lpTcp = ( tcp_impl * )lpAioKey->m_pHandle;
        socket_impl * lpSocket = &lpTcp->m_socket;
        SOCKET hSocket = lpTcp->m_socket.m_hSocket;
        buf_impl * lpBuf = 0;

        lpAioKey->m_nRequestNum++;

        SOCKET hClientSocket = INVALID_SOCKET;
        aio_data * lpRetData = 0;

        // 
        // prepare for result value
        // 
        {
            lpRetData = m_pAioImpl->m_data_pool.GetItem();
            lpRetData->init();
            lpRetData->m_eAioType = i_aio::AIO_TCP_ACCPET_RET;
            lpRetData->m_lpMsgData = lpMsg;
            lpMsg->m_nData[0] = lpRetData->m_eAioType;
        }

        if( 0 == nRet )
        {
            hClientSocket = ::WSASocket( AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED );

            if( INVALID_SOCKET == hClientSocket )
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            tcp_impl * pTcp = 0;
            g_pXosImpl->create( i_xos::XOS_OBJ_TCP, ( void ** )&pTcp );
            socket_impl * pSocket = &pTcp->m_socket;
            pSocket->m_hSocket = hClientSocket;

            aio_key * pKey = m_pAioImpl->m_key_pool.GetItem();
            pKey->init();
            pKey->m_lpUserKey = lpUserKey;
            pKey->m_pHandle = pTcp;

            lpMsg->m_lpData[2] = pKey;
        }

        if( 0 == nRet )
        {
            nRet = g_pXosImpl->create( i_xos::XOS_OBJ_BUF, ( void ** )&lpBuf );
            lpMsg->set_buf( 0, lpBuf );
        }

        if( 0 == nRet )
        {
            DWORD dwSize = sizeof( sockaddr_in ) + 16;
            lpMsg->m_bData[0] = true;

            if( !m_pAioImpl->m_lpAcceptEx( hSocket, hClientSocket, lpBuf->m_szBuf, 0, dwSize, dwSize, &lpRetData->m_dwBytes, ( WSAOVERLAPPED * )&lpRetData->m_ov_ex ) )
            {
                DWORD dwError = WSAGetLastError();

                if( ERROR_IO_PENDING != dwError )
                {
                    nRet = 1;
                }
            }
        }

        if( 0 != nRet )
        {
            lpMsg->m_bData[0] = false;
            PostQueuedCompletionStatus( m_pAioImpl->m_hIocp, 0, 0, ( WSAOVERLAPPED * )&lpRetData->m_ov_ex );
        }

        lpRetData = 0;

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : recv len
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_recv_ret( int nResult, DWORD dwBytes, aio_key * lpKey, aio_data * lpData )
    {
        using namespace std;
        int nRet = 0;

        msg_impl * lpMsg = lpData->m_lpMsgData;
        aio_key * lpAioKey = ( aio_key * )lpMsg->m_lpData[1];
        buf_impl * lpBuf = lpMsg->m_lpBuf[0];

         --lpAioKey->m_nRequestNum;

        if( !lpMsg->m_bData[0] || ( 0 != nResult ) || ( 0 == dwBytes ) )
        {
            nRet = 1;
        }

        if( 0 == nRet )
        {
            lpBuf->m_szBuf[dwBytes] = 0;
            lpBuf->m_nLen = dwBytes;
            lpMsg->m_nData[1] = dwBytes;
        }

        if( 0 != nRet )
        {
            lpMsg->m_bData[0] = false;
        }

        //cout << " internal read request num =" << lpAioKey->m_nRequestNum << endl;
        m_pAioImpl->m_pCallbackThread->notify( lpMsg );

        ///////////////////////////////////////////////////////////////////////////////////////////////
        // 
        // close connect
        // 

        if( ( aio_key::KEY_STATE_CLOSING == lpAioKey->m_eState ) && ( 0 == lpAioKey->m_nRequestNum ) )
        {
            helper_post_close( lpAioKey );
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : recv len
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_recv( int nResult, DWORD dwBytes, aio_key * lpKey, aio_data * lpData )
    {
        int nRet = 0;

        msg_impl * lpMsg = lpData->m_lpMsgData;
        void * lpUserKey = lpMsg->m_lpData[0];
        aio_key * lpAioKey = ( aio_key * )lpMsg->m_lpData[1];
        tcp_impl * lpTcp = ( tcp_impl * )lpAioKey->m_pHandle;
        socket_impl * lpSocket = &lpTcp->m_socket;
        SOCKET hSocket = lpTcp->m_socket.m_hSocket;
        aio_data * lpRetData = 0;
        buf_impl * lpBuf = lpMsg->m_lpBuf[0];

        lpAioKey->m_nRequestNum++;

        if( 0 == nRet )
        {
            lpRetData = m_pAioImpl->m_data_pool.GetItem();
            lpRetData->init();
            lpRetData->m_eAioType = i_aio::AIO_TCP_RECV_RET;
            lpRetData->m_lpMsgData = lpMsg;
            lpMsg->m_nData[0] = lpRetData->m_eAioType;
        }

        if( 0 == nRet )
        {
            lpMsg->m_bData[0] = true;

            int ret = ::WSARecv( hSocket, &lpBuf->m_wsa_buf, 1, &lpRetData->m_dwBytes, &lpRetData->m_dwFlags, ( WSAOVERLAPPED * )&lpRetData->m_ov_ex, 0 );

            if( SOCKET_ERROR == ret )
            {
                DWORD dwErr = WSAGetLastError();

                if( WSA_IO_PENDING != dwErr )
                {
                    nRet = 1;
                }
            }
        }

        if( 0 != nRet )
        {
            lpMsg->m_bData[0] = false;
            PostQueuedCompletionStatus( m_pAioImpl->m_hIocp, 0, 0, ( WSAOVERLAPPED * )&lpRetData->m_ov_ex );
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : send bytes
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_send_ret( int nResult, DWORD dwBytes, aio_key * lpKey, aio_data * lpData )
    {
        using namespace std;
        int nRet = 0;

        msg_impl * lpMsg = lpData->m_lpMsgData;
        aio_key * lpAioKey = ( aio_key * )lpMsg->m_lpData[1];
        buf_impl * lpBuf = lpMsg->m_lpBuf[0];

        --lpAioKey->m_nRequestNum;

        if( !lpMsg->m_bData[0] || ( 0 != nResult ) )
        {
            nRet = 1;
        }

        if( 0 == nRet )
        {
            lpMsg->m_nData[1] = dwBytes;
        }

        if( 0 != nRet )
        {
            lpMsg->m_bData[0] = false;
        }

        //cout << " internal send request num =" << lpAioKey->m_nRequestNum << endl;
        m_pAioImpl->m_pCallbackThread->notify( lpMsg );

        ///////////////////////////////////////////////////////////////////////////////////////////////
        // 
        // close connect
        // 

        if( ( aio_key::KEY_STATE_CLOSING == lpAioKey->m_eState ) && ( 0 == lpAioKey->m_nRequestNum ) )
        {
            helper_post_close( lpAioKey );
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : send bytes
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_send( int nResult, DWORD dwBytes, aio_key * lpKey, aio_data * lpData )
    {
        int nRet = 0;

        msg_impl * lpMsg = lpData->m_lpMsgData;
        void * lpUserKey = lpMsg->m_lpData[0];
        aio_key * lpAioKey = ( aio_key * )lpMsg->m_lpData[1];
        tcp_impl * lpTcp = ( tcp_impl * )lpAioKey->m_pHandle;
        socket_impl * lpSocket = &lpTcp->m_socket;
        SOCKET hSocket = lpTcp->m_socket.m_hSocket;
        aio_data * lpRetData = 0;
        buf_impl * lpBuf = lpMsg->m_lpBuf[0];

        lpAioKey->m_nRequestNum++;

        if( 0 == nRet )
        {
            lpRetData = m_pAioImpl->m_data_pool.GetItem();
            lpRetData->init();
            lpRetData->m_eAioType = i_aio::AIO_TCP_SEND_RET;
            lpRetData->m_lpMsgData = lpMsg;
            lpMsg->m_nData[0] = lpRetData->m_eAioType;
        }

        if( 0 == nRet )
        {
            lpBuf->m_wsa_buf.len = lpBuf->m_nLen;
            lpMsg->m_bData[0] = true;

            int ret = ::WSASend( hSocket, &lpBuf->m_wsa_buf, 1, &lpRetData->m_dwBytes, lpRetData->m_dwFlags, ( WSAOVERLAPPED * )&lpRetData->m_ov_ex, 0 );

            if( SOCKET_ERROR == ret )
            {
                DWORD dwErr = WSAGetLastError();

                if( WSA_IO_PENDING != dwErr )
                {
                    nRet = 1;
                }
            }
        }

        if( 0 != nRet )
        {
            lpMsg->m_bData[0] = false;
            PostQueuedCompletionStatus( m_pAioImpl->m_hIocp, 0, 0, ( WSAOVERLAPPED * )&lpRetData->m_ov_ex );
        }

        return nRet;
    }

    // 
    // m_nData[0] : cmd type
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::close_handle_ret( int nResult, DWORD dwBytes, aio_key * lpKey, aio_data * lpData )
    {
        int nRet = 0;

        msg_impl * lpMsg = lpData->m_lpMsgData;
        aio_key * lpAioKey = ( aio_key * )lpMsg->m_lpData[1];

        if( lpAioKey )
        {
            lpMsg->m_lpData[1] = 0;
            lpAioKey->release();
            lpAioKey = 0;
        }

        m_pAioImpl->m_pCallbackThread->notify( lpMsg );

        return nRet;
    }

    // 
    // m_nData[0] : cmd type
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::close_handle( int nResult, DWORD dwBytes, aio_key * lpKey, aio_data * lpData )
    {
        int nRet = 0;

        msg_impl * lpMsg = lpData->m_lpMsgData;
        void * lpUserKey = lpMsg->m_lpData[0];
        aio_key * lpAioKey = ( aio_key * )lpMsg->m_lpData[1];
        i_handle * lpHandle = lpAioKey->m_pHandle;

        lpAioKey->m_eState = aio_key::KEY_STATE_CLOSING;

        if( lpHandle )
        {
            lpHandle->close();
        }

        if( 0 == lpAioKey->m_nRequestNum )
        {
            aio_data * lpRetData = m_pAioImpl->m_data_pool.GetItem();
            lpRetData->init();
            lpRetData->m_eAioType = i_aio::AIO_TCP_CLOSE_HANDLE_RET;
            lpRetData->m_lpMsgData = lpMsg;
            lpMsg->m_nData[0] = lpRetData->m_eAioType;
            lpMsg->m_bData[0] = true;
            PostQueuedCompletionStatus( m_pAioImpl->m_hIocp, 0, 0, ( WSAOVERLAPPED * )&lpRetData->m_ov_ex );
            lpRetData = 0;
            lpMsg = 0;
        }

        if( lpMsg )
        {
            lpMsg->release();
            lpMsg = 0;
        }

        return nRet;
    }

    // 
    // m_nData[0] : cmd type
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::helper_post_close( aio_key * lpKey )
    {
        int nRet = 0;

        aio_data * lpData = 0;
        msg_impl * lpMsg = 0;

        if( 0 == nRet )
        {
            nRet = g_pXosImpl->create( i_xos::XOS_OBJ_MSG, ( void ** )&lpMsg );
        }

        if( 0 == nRet )
        {
            lpMsg->m_nData[0] = i_aio::AIO_TCP_CLOSE_HANDLE_RET;
            lpMsg->m_lpData[0] = lpKey->m_lpUserKey;
            lpMsg->m_bData[0] = true;
            lpData = m_pAioImpl->m_data_pool.GetItem();
            lpData->init();
        }

        if( 0 == nRet )
        {
            lpData->m_eAioType = ( i_aio::enumAioType )lpMsg->m_nData[0];
            lpData->m_lpMsgData = lpMsg;
        }

        if( 0 == nRet )
        {
            m_pAioImpl->m_pCallbackThread->notify( lpMsg );
            lpMsg = 0;
            lpKey->release();
            lpKey = 0;
        }

        return nRet;
    }

} // xos
